use flutter_rust_bridge::{frb, DartFnFuture};
use futures::stream::{StreamExt, TryStreamExt};
use indicatif::HumanDuration;
use log::info;
use reqwest::Url;
use std::collections::hash_map::DefaultHasher;
use std::fmt::Write;
use std::hash::{Hash, Hasher};
use std::path::{Path, PathBuf};
use std::time::Instant;
use tokio::fs::{self, File};
use tokio::io::AsyncWriteExt;
use tokio::process::Command;
use tokio::sync::mpsc;

use crate::encrypt;
/// 最大同时下载数
static MAX_PARALLEL_DOWNLOAD: usize = 20;
static M3U8_LIST_PATH: &str = "input.txt";

type Result<Output> = anyhow::Result<Output>;

#[derive(Debug, Clone, Copy, PartialEq)]
pub enum M3u8TaskStatus {
    Pending,     // 待下载
    Downloading, // 正在下载
    Paused,      // 下载暂停
    Failed,      // 下载失败
    Completed,   // 下载完成
    Cancelled,   // 已取消
}

#[derive(Debug)]
pub struct M3u8Task {
    pub url: String,
    pub status: M3u8TaskStatus,
    pub worker: usize,
    pub base_url: String,
    pub work_dir: String,
    pub cache_dir: String,
    pub name: String,
    pub m3u8_path: String,
    pub current_index: u64,
    key: Option<Vec<u8>>,
    client: reqwest::Client,
    ts_list: Vec<String>,
    pub total: u64,
    pub consume: u64,
}

impl M3u8Task {
    #[frb(sync)]
    pub fn new(url: String, work_dir: String, name: String, work: u64) -> Self {
        Self {
            url,
            work_dir,
            name,
            key: None,
            worker: work as usize,
            base_url: "".to_string(),
            cache_dir: "".to_string(),
            m3u8_path: "".to_string(),
            client: reqwest::ClientBuilder::new()
                .use_rustls_tls()
                .build()
                .unwrap(),
            consume: 0,
            current_index: 0,
            total: 0,
            ts_list: Vec::with_capacity(0),
            status: M3u8TaskStatus::Pending,
        }
    }
    pub async fn init(&mut self) -> anyhow::Result<()> {
        let url = self.url.as_str();
        self.base_url = url
            .split_at(url.rfind('/').expect("please input the m3u8 url"))
            .0
            .to_string();
        // 生成临时下载目录
        self.cache_dir = make_sure_url_dir(url, &self.work_dir)
            .await?
            .as_ref()
            .to_str()
            .unwrap()
            .to_string();

        // 下载文件清单文件
        self.m3u8_path = format!("{}/{}", self.cache_dir, M3U8_LIST_PATH);

        let (base_url, ts_list_cotent_origin) = resolve_m3u8_list(&self.client, &self.url).await?;

        self.base_url = base_url;

        let (m3u8, key) = self.download_key(ts_list_cotent_origin).await?;

        self.key = key;
        let mut m3u8_list: Vec<String> = m3u8
            .lines()
            .map(|line| line.trim_start_matches("/").to_string())
            .collect();

        m3u8_list = remove_ad(m3u8_list).await;
        let m3u8 = m3u8_list
            .iter()
            .filter(|line| !line.starts_with("#") && line.len() > 2)
            .fold(String::new(), |mut buf, line| {
                // if line.starts_with("#") {
                //     let _ = writeln!(buf, "{}", line);
                // } else {
                // let _ = writeln!(buf, "{}/{}", self.cache_dir, line);
                // }
                let _ = writeln!(buf, "file {}", line);
                buf
            });

        if !PathBuf::from(&self.m3u8_path).exists() {
            let mut tmp_list_file = File::create(&self.m3u8_path).await?;
            tmp_list_file.write_all(m3u8.as_bytes()).await?;
        }

        self.ts_list = m3u8_list
            .into_iter()
            .filter(|line| !line.is_empty() && !line.trim_start().starts_with('#'))
            .collect();

        self.total = self.ts_list.len() as u64;

        Ok(())
    }

    pub async fn download(
        &mut self,
        callback: impl Fn(f32) -> DartFnFuture<()>,
    ) -> anyhow::Result<()> {
        info!("download ts start");
        let started = Instant::now();
        let worker = std::cmp::min(self.worker, MAX_PARALLEL_DOWNLOAD);

        let cache_dir = self.cache_dir.clone();
        self.status = M3u8TaskStatus::Downloading;
        let client = self.client.clone();
        let ts_list = self.ts_list.clone();
        let base_url = self.base_url.clone();
        let key = self.key.clone();
        let (tx, mut rx) = mpsc::channel::<usize>(worker);
        info!("base url: {}", base_url);
        info!("总计: {}", ts_list.len());
        info!("总计: {:?}", ts_list);
        tokio::spawn(async move {
            match futures::stream::iter(ts_list)
                .enumerate()
                .map(|(index, ts)| {
                    let client = client.clone();
                    let ts_url = format!("{}/{}", base_url, ts);
                    let ts_file_path = PathBuf::from(&cache_dir).join(ts.trim_start_matches("/"));
                    let tx = tx.clone();
                    let key = key.clone();
                    async move {
                        download_ts(tx.clone(), client, index, ts_url, ts_file_path, key).await?;
                        Ok(()) as Result<()>
                    }
                })
                .buffer_unordered(worker)
                .try_collect::<()>()
                .await
            {
                Ok(_) => {
                    // 打印用时统计
                    info!(
                        "take: {}",
                        HumanDuration(Instant::now().duration_since(started))
                    );
                }
                Err(e) => {
                    info!("{:?}", e);
                }
            }
        });
        info!("async download callback");
        while let Some(_) = rx.recv().await {
            self.current_index += 1;
            callback(self.current_index as f32 / self.total as f32).await
        }
        self.status = M3u8TaskStatus::Completed;

        Ok(())
    }

    async fn download_key(&mut self, m3u8: String) -> anyhow::Result<(String, Option<Vec<u8>>)> {
        let mut key_bytes = None;
        if m3u8.contains("#EXT-X-KEY") {
            let mut new = m3u8.clone();
            let key = m3u8
                .lines()
                .filter(|line| !line.is_empty() && line.starts_with("#EXT-X-KEY"))
                .next()
                .unwrap();

            // 替换 key
            if let Some((a, mut b)) = key.split_once("=\"") {
                info!("{} {}", a, b);
                if b.starts_with("/") {
                    let old = b.to_string();
                    b = b.trim_start_matches("/");
                    new = m3u8.replace(&old, &(format!("{}/{}", self.cache_dir, b)));
                    b = b.trim_end_matches("\"")
                }
                let key_file_path = PathBuf::from(&self.cache_dir).join(b);
                if !key_file_path.exists() {
                    let key_url = format!("{}/{}", self.base_url, b);
                    let response = self.client.get(key_url).send().await?;
                    let out_data = response.bytes().await?;
                    key_bytes = Some(out_data.clone().to_vec());
                    // 获取文件的父目录路径
                    let dir_path = key_file_path.parent().ok_or_else(|| {
                        anyhow::format_err!("No parent directory found for the file path")
                    })?;

                    // 创建父目录（包括所有必要的中间目录）
                    fs::create_dir_all(dir_path).await?;

                    let mut out_file = File::create(key_file_path).await?;
                    out_file.write_all(&out_data).await?;
                }
            }
            return Ok((new, key_bytes));
        }
        Ok((m3u8, key_bytes))
    }
}

async fn download_ts(
    tx: mpsc::Sender<usize>,
    client: reqwest::Client,
    index: usize,
    url: String,
    dest: PathBuf,
    key: Option<Vec<u8>>,
) -> Result<()> {
    if dest.exists() {
        tx.send(index).await?;
        return Ok(());
    }
    // https://gist.github.com/giuliano-oliveira/4d11d6b3bb003dba3a1b53f43d81b30d
    let response = client.get(url).send().await?;

    let total_size = response
        .content_length()
        .ok_or_else(|| anyhow::format_err!("can't get the content_length"))?;

    let part_path: PathBuf = format!(
        "{}{}",
        dest.to_str()
            .ok_or_else(|| anyhow::format_err!("dest({:?}) to_str error", dest))?,
        ".part"
    )
    .parse()
    .unwrap();
    // 获取文件的父目录路径
    let dir_path = part_path
        .parent()
        .ok_or_else(|| anyhow::format_err!("No parent directory found for the file path"))?;

    // 创建父目录（包括所有必要的中间目录）
    fs::create_dir_all(dir_path).await?;

    let mut part = File::create(&part_path).await?;
    let mut downloaded: u64 = 0;

    // 流下载
    let mut stream = response.bytes_stream();
    while let Some(item) = stream.next().await {
        let chunk = item?;

        if let Some(ref key) = key {
            let mut data = chunk.to_vec();
            let key = key.to_vec();
            let c = encrypt::dec_aes_128(&mut data, &key, &key).unwrap();
            part.write_all(c).await?;
        } else {
            part.write_all(&chunk).await?;
        }

        downloaded = std::cmp::min(downloaded + (chunk.len() as u64), total_size);
    }

    fs::rename(part_path, dest).await?;
    tx.send(index).await?;
    info!("download ts {} ok", index);
    Ok(())
}

#[allow(unused)]
async fn merge_video<P1, P2>(tmp_dir: P1, dest: P2) -> Result<()>
where
    P1: AsRef<Path>,
    P2: AsRef<std::ffi::OsStr>,
{
    // 调用 ffmpeg 合并文件
    // command = 'ffmpeg -y -f concat -i %s -bsf:a aac_adtstoasc -c copy %s' % (concatfile, path)
    // Command::new("ffmpeg")
    //     .arg("-y")
    //     .arg("-f")
    //     .arg("concat")
    //     .arg("-i")
    //     .arg(M3U8_LIST_PATH)
    //     .arg("-bsf:a")
    //     .arg("aac_adtstoasc")
    //     .arg("-c")
    //     .arg("copy")
    //     .arg(dest)
    //     .current_dir(tmp_dir)
    //     .spawn()?
    //     .wait()
    //     .await?;

    // ffmpeg 处理加密ts
    // ffmpeg -y -allowed_extensions ALL -protocol_whitelist "file,http,crypto,tcp,https" -i ts_list.m3u8 -bsf:a aac_adtstoasc  -c copy output.mp4

    Command::new("ffmpeg")
        .arg("-y")
        .arg("-allowed_extensions")
        .arg("ALL")
        .arg("-protocol_whitelist")
        .arg("file,http,crypto,tcp,https")
        .arg("-i")
        .arg(M3U8_LIST_PATH)
        .arg("-bsf:a")
        .arg("aac_adtstoasc")
        .arg("-c")
        .arg("copy")
        .arg(dest)
        .current_dir(tmp_dir)
        .spawn()?
        .wait()
        .await?;

    Ok(())
}

async fn make_sure_url_dir(url: &str, work_dir: &str) -> Result<impl AsRef<Path>> {
    // url hash
    let mut hasher = DefaultHasher::new();
    url.hash(&mut hasher);
    let url_hash = hasher.finish();

    // create cache_dir
    let url_dir = PathBuf::from(work_dir).join(url_hash.to_string());
    // let url_dir = ProjectDirs::from("", "", "m3u8d")
    //     .ok_or_else(|| anyhow::anyhow!("not find ProjectDirs"))?
    //     .cache_dir()
    //     .join(url_hash.to_string());
    info!("use cache dir: {}", url_dir.to_string_lossy());
    make_sure_dir_exsit(url_dir).await
}

// AsRef::<Path>::as_ref("/tmp");
// async fn make_sure_dir_exsit(path: impl AsRef<Path>) -> Result<impl AsRef<Path>> {
async fn make_sure_dir_exsit<P: AsRef<Path>>(path: P) -> Result<impl AsRef<Path>> {
    if path.as_ref().exists() {
        return Ok(path);
    }
    fs::create_dir_all(path.as_ref()).await?;
    Ok(path)
}

async fn remove_ad(m3u8_list: Vec<String>) -> Vec<String> {
    let mut ad = false;

    let mut new_m3u8_list = Vec::new();

    for (i, e) in m3u8_list.iter().enumerate() {
        if e.eq("#EXT-X-DISCONTINUITY") {
            if i < m3u8_list.len() - 2
                && !m3u8_list[i - 1].starts_with("#")
                && m3u8_list[i + 1].starts_with("#")
                && m3u8_list[i - 1][0..1] != m3u8_list[i + 2][0..1]
            {
                ad = !ad;
            }
            continue;
        }
        if !ad && e.len() > 0 {
            new_m3u8_list.push(e.clone());
        }
    }

    new_m3u8_list
}

async fn resolve_m3u8_list(
    client: &reqwest::Client,
    url: &str,
) -> anyhow::Result<(String, String), reqwest::Error> {
    info!("resolve {}", url);
    let text = client.get(url).send().await?.text().await?;
    if !text.contains("#EXT-X-STREAM-INF") {
        if text
            .lines()
            .filter(|line| line.starts_with("/"))
            .next()
            .is_some()
        {
            if let Ok(url) = Url::parse(url) {
                let mut base_url = format!("{}://{}", url.scheme(), url.domain().unwrap());
                if let Some(port) = url.port() {
                    base_url = format!("{}:{}", base_url, port)
                }
                return Ok((base_url, text));
            }
        }

        let base_url = url
            .split_at(url.rfind('/').expect("please input the m3u8 url"))
            .0;
        return Ok((base_url.to_string(), text));
    }
    let last = text.lines().last().unwrap();
    let base_url = url
        .split_at(url.rfind('/').expect("please input the m3u8 url"))
        .0;
    let url = format!("{}/{}", base_url, last);
    return Box::pin(resolve_m3u8_list(client, &url)).await;
}
